package xt3FuturesUser

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"sync"
	"time"
	"wscoin/core"
	"wscoin/initialize"

	"github.com/astaxie/beego/logs"
	"github.com/batchcorp/rabbit"
	redis "github.com/go-redis/redis/v9"
	"github.com/gorilla/websocket"
	"github.com/recws-org/recws"
	"github.com/streadway/amqp"
	"github.com/wxnacy/wgo/arrays"
)

const (
	// redis Key Get all market
	redisKey   = "xtv3userfuture:markets"
	tradeLimit = 1000

	topicUser   = "sub_user"
	subscribe   = "SUBSCRIBE"
	unsubscribe = "UNSUBSCRIBE"

	subUserBalance  = "user.balance"
	subUserPosition = "user.position"
	subUserTrade    = "user.trade"
	subUserOrder    = "user.order"
	subUserNotify   = "user.notify"

	// MQ-name
	queueName         = "xtv3userfuture:%s:%s"
	exchangeName      = "xtv3userfuture" // exchange Name
	exchangeNameRedis = "xtv3userfuture"
)

var (
	// redis Key Get susub market Or blacklist
	unSubRedisKey = "xtv3userfuture:markets:blacklist"
	// redis Key Get recover market
	recoverRedisKey = "xtv3userfuture:markets:recover:%d"
	// Own subscription to the market information
	ownerRedisKey = "xtv3userfuture:markets:sub:%d"
)

type event struct {
	Channel   string   `json:"req"`
	ListenKey string   `json:"listenKey"`
	Userinfo  *AuthKey `json:"-"`
}

func (e *event) MarshalJSON() ([]byte, error) {

	xtAPI := &XTContractListkey{}
	listkey := xtAPI.Request(e.Userinfo.Accesskey, e.Userinfo.SecretKey)
	fmt.Println("[wsContractUser-Event] :", listkey, e.Userinfo.Accesskey)
	return json.Marshal(struct {
		Channel   string `json:"req"`
		ListenKey string `json:"listenKey"`
	}{
		Channel:   e.Channel,
		ListenKey: listkey,
	})
}

type XTContractUserEvent struct {
	Url   string `json:"-"`
	Event []*event
}

func NewXTContractUserEvent() *XTContractUserEvent {
	return &XTContractUserEvent{
		Url:   baseURL,
		Event: make([]*event, 0),
	}
}

// onevent
func (x *XTContractUserEvent) OnEvent(userinfo *AuthKey) {

	// xtAPI := &XTContractListkey{}
	// listkey := xtAPI.Request(userinfo.Accesskey, userinfo.SecretKey)
	// x.Event = append(x.Event, &event{Channel: topicUser, ListenKey: listkey, Userinfo: userinfo})
	x.Event = append(x.Event, &event{Channel: topicUser, Userinfo: userinfo})
}

// outevent
func (x *XTContractUserEvent) OutEvent() {
	// TODO

}

func (b *XTContractUserEvent) PingMessageHandler(w *websocket.Conn) error {
	return w.WriteMessage(websocket.TextMessage, []byte("ping"))
}

func (b *XTContractUserEvent) GetURI() (uri string) {
	uri = b.Url
	return
}

type AuthKey struct {
	Accesskey string `json:"accesskey"`
	SecretKey string `json:"secretkey"`
}

type Manager struct {
	UserInfo    *AuthKey // Accesskey
	WSClient    *recws.RecConn
	MQClientMgr map[string]*rabbit.Rabbit
	RecvMessage func(msg []byte)

	mu     sync.RWMutex
	ctx    context.Context
	cancel context.CancelFunc
}

func NewManager(userInfo *AuthKey) *Manager {

	ctx, cancel := context.WithCancel(context.Background())
	return &Manager{
		UserInfo:    userInfo,
		ctx:         ctx,
		cancel:      cancel,
		MQClientMgr: make(map[string]*rabbit.Rabbit),
	}
}

func (m *Manager) AddMQ(name string, session *rabbit.Rabbit) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.MQClientMgr[name] = session
}

func (m *Manager) GetMQ(name string) (*rabbit.Rabbit, bool) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	session, ok := m.MQClientMgr[name]
	return session, ok
}

func (m *Manager) initMQ(cfg *initialize.Configure) {

	// MQ建立
	userMQName := fmt.Sprintf(queueName, topicUser, m.UserInfo.Accesskey)
	rabbitOpt := initialize.NewRabbitOption(userMQName, exchangeNameRedis)
	client := cfg.RabbitMgr.GetClient(rabbitOpt)
	m.AddMQ(m.UserInfo.Accesskey, client)

}

func (m *Manager) initWs() {
	// WS建立
	event := NewXTContractUserEvent()

	event.OnEvent(m.UserInfo)

	m.WSClient = &recws.RecConn{
		SubscribeHandler: func() error {
			var err error
			for _, topic := range event.Event {

				err = m.WSClient.WriteJSON(topic)
			}
			return err
		},
	}
	m.WSClient.Dial(event.GetURI(), http.Header{})
}

func (m *Manager) getMessage(event string, message []byte, f func(string, []byte)) {
	f(event, message)
	return
}

// consumer
// 消费MQ数据写入到缓存中
// block=True
func (m *Manager) dispatchMsg(cfg *initialize.Configure) {

	// depth
	clientMQ, ok := m.GetMQ(m.UserInfo.Accesskey)
	if ok {
		client := cfg.RedisMgr.GetClient(0)
		depthHandle := NewEventHandle(
			clientMQ,
			clientMQ.Options.QueueName,
			m.UserInfo,
			client,
			m.ctx,
		)
		go clientMQ.Consume(m.ctx, nil, depthHandle.DepthHandleConsumer)
	}

}

// 循环读取ReadMessage
// block=True
func (m *Manager) recvMsg() {

	go func() {
		ticker := time.NewTicker(time.Second * 10)
		defer ticker.Stop()
		RestWsSub := time.NewTicker(time.Hour * 4)
		defer RestWsSub.Stop()
		for {

			select {
			case <-m.ctx.Done():
				logs.Info("[xt_contract_user-run] ReadMessage exit ...", m.UserInfo)
				return
			case <-ticker.C:
				//
				if err := m.WSClient.WriteMessage(websocket.TextMessage, []byte("ping")); err != nil {
					log.Printf("Error: WriteMessage %s  err: %s  market: %s", m.WSClient.GetURL(), err.Error(), m.UserInfo.Accesskey)
					m.WSClient.CloseAndReconnect()
					continue
				}

			case <-RestWsSub.C:
				//
				err := m.WSClient.SubscribeHandler()
				logs.Debug("[xt_contract_user-run] SubscribeHandler err ", err)
			default:
				//
				_, msg, err := m.WSClient.ReadMessage()
				if !m.WSClient.IsConnected() {
					time.Sleep(3 * time.Second)
					continue
				}

				if err != nil {
					logs.Info("[xt_contract_user-run] ReadMessage ", err, m.UserInfo, string(msg))
					time.Sleep(3 * time.Second)
					continue
				}

				topic := NewRecvMessage(msg)
				if topic.IsValid() {
					continue
				}

				handle := func(quant string, msgByte []byte) {

					mqQuant, _ := m.GetMQ(quant)
					// fmt.Println("[ReadMessage] ", quant, string(msg))
					if err := mqQuant.Publish(
						m.ctx,
						mqQuant.Options.QueueName, msgByte); err != nil {
						fmt.Printf("[xt_contract_user-run] Push failed: %s\n", err)
					}
				}
				m.getMessage(m.UserInfo.Accesskey, msg, handle)

			}

		}
	}()

}

// 启动
func (m *Manager) Start(cfg *initialize.Configure) {

	m.initMQ(cfg)
	m.initWs()
	m.dispatchMsg(cfg)
	m.recvMsg()

}

// 关闭所有的MQ连接
func (m *Manager) Stop() {
	for _, client := range m.MQClientMgr {
		client.Close()
	}
	m.cancel()
}

/*
 * 处理对应事件的数据消费动作
 *
 */
type EventHandle struct {
	ctx         context.Context
	client      *rabbit.Rabbit
	depth       *CacheDepth
	userInfo    *AuthKey
	quantChQ    string
	redisClient *redis.Client
}

func NewEventHandle(
	client *rabbit.Rabbit,
	quantChQ string,
	userInfo *AuthKey,
	redisClient *redis.Client,
	ctx context.Context,
) *EventHandle {
	return &EventHandle{
		client:      client,
		quantChQ:    quantChQ,
		userInfo:    userInfo,
		ctx:         ctx,
		depth:       &CacheDepth{},
		redisClient: redisClient,
	}
}

func (e EventHandle) ClearDepthLimitLoop() {
	e.depth.ResetLimit(e.ctx)
}

func (e EventHandle) DepthHandleConsumer(msg amqp.Delivery) error {

	select {
	case <-e.ctx.Done():
		logs.Info("[xt_contract_user-DepthHandleConsumer] ReadMessage exit ...", e.userInfo)
		e.client.Stop()

	default:

		content, err := UserInfoConsumerHandle(
			msg.Body,
		)
		if err != nil {
			logs.Info("[xt_contract_user-DepthHandleConsumer] Unmarshal model.RecvTopic", string(msg.Body))
		}
		userPush := e.quantChQ + ":" + content.Channel + ":push"
		e.redisClient.Publish(context.TODO(), userPush, Strval(content.Data))

		userPushPing := e.quantChQ + ":" + "ping"
		nt := time.Now().UnixMilli()
		e.redisClient.Set(context.TODO(), userPushPing, nt, 0)

	}
	return nil
}

// 入口
type Engine struct {
	managers    map[string]*Manager
	cfg         *initialize.Configure
	redisClient *core.RedisOperate
	start, end  int
}

func NewEngine(start, end int) *Engine {
	recoverRedisKey = fmt.Sprintf(recoverRedisKey, start)
	ownerRedisKey = fmt.Sprintf(ownerRedisKey, start)
	return &Engine{
		managers: make(map[string]*Manager),
		start:    start,
		end:      end,
	}
}

// 加载所有匹配的市场
func (e *Engine) Start(cfg *initialize.Configure) {

	e.cfg = cfg
	client := cfg.RedisMgr.GetClient(0)
	e.redisClient = core.NewRedisOperateClient(client)
	e.redisClient.AddSubExchangeKeys(recoverRedisKey)
	_, origins := e.redisClient.AcquireOwnerMarketInfoFromCache(ownerRedisKey)
	markets := origins
	// e.load(markets[:1], origins[:1])
	e.load(markets, origins)
	go e.loopLoadCancelMakret()
	e.loopLoadRecoverMakret()
}

func (e *Engine) load(markets, origins []string) {

	for i := 0; i < len(markets); i++ {

		if e.isContainsString(origins[i]) {
			continue
		}

		userInfo := new(AuthKey)
		err := json.Unmarshal([]byte(markets[i]), userInfo)
		if err != nil {
			logs.Info("[run-xt-contract-user-Unmarshal] err: ", err)
		}
		manager := NewManager(userInfo)
		e.managers[origins[i]] = manager
		manager.Start(e.cfg)
		time.Sleep(time.Millisecond * 100)

	}
}

func (e *Engine) close(origins []string) {

	// 清理自身订阅的市场key中用于管理自身订阅的市场信息
	e.redisClient.Client.SRem(context.TODO(), ownerRedisKey, origins)
	for _, symbol := range origins {
		if manager, ok := e.managers[symbol]; ok {
			manager.Stop()
		}
	}

}

// 移除blacklist市场
func (e Engine) isContainsString(symbol string) bool {
	_, origins := e.redisClient.AcquireOwnerMarketInfoFromCache(unSubRedisKey)
	return arrays.ContainsString(origins, symbol) != -1
}

// 加载恢复市场
func (e *Engine) loopLoadRecoverMakret() {
	for {
		markets, origins := e.redisClient.AcquireInfoFromCache(recoverRedisKey)
		// 写入到自身订阅的市场key中用于管理自身订阅的市场信息
		e.redisClient.Client.SAdd(context.TODO(), ownerRedisKey, origins)
		e.redisClient.Client.ZRem(context.TODO(), recoverRedisKey, origins)

		members := make([]redis.Z, 0)
		for data := range origins {
			members = append(members, redis.Z{
				Score:  1,
				Member: data,
			})
		}

		e.redisClient.Client.ZAdd(context.TODO(), redisKey, members...)

		e.load(markets, origins)
		time.Sleep(time.Second)
	}
}

// 取消的市场
func (e *Engine) loopLoadCancelMakret() {
	for {
		_, origins := e.redisClient.AcquireOwnerMarketInfoFromCache(unSubRedisKey)
		e.close(origins)
		time.Sleep(time.Second)
	}
}
